package raft

import (
	"edata-hub/pkg/raft/services"
	"github.com/gin-contrib/cors"
	"github.com/gin-gonic/gin"
	"github.com/gorilla/websocket"
	"google.golang.org/grpc"
	"net"
	"net/http"
	"sync"
	"time"
)

// Raft
/*Raft主体结构*/
type Raft struct {
	/*----------------------------------------------------------------------------------------------------------------*/
	//  设置启动横幅
	Banner string
	//  当前 Raft Server 的 IP:Port
	Addr string
	//   当前 Raft Server 的角色，Leader，Candidate，Follower
	Role string
	//   Master集群信息，key代表地址，value代表Leader 所记录的 Follower 的最新日志 index
	//               该index一旦被记录，说明当前index以及之前所记录的日志是没问题的。
	Peers map[string]Peer
	//  当前Worker集群的信息，key是worker地址
	Workers map[string]WorkerInfo
	//   监控节点的信息，map[string]MonitorStandard结构的key是时间
	MonitorInfo map[string]map[string]string
	//   该Raft Server 所标记的任期
	Term int64
	//   该 Raft Server 所标记的 Leader
	Leader string
	//   Raft日志
	Log *Logs
	// 输出日志
	Msg *Message
	//   Raft 配置对象
	Config RaftConfig
	//   //持久化的日志
	PersistLogs []byte
	/*----------------------------------------------------------------------------------------------------------------*/
	//命令名称，命令对应执行方法
	Commands map[string]func(args map[string]string) CommandExecRet
	//命令优化器，接收到的命令会先经过优化，二次开发可以自定义该方法
	CommandOptimizer func(entry CommandEntry) ([]CommandEntry, error)
	//命令名称，命令对应描述
	CommandsDes map[string]string
	//命令队列
	//CommandQueues map[string][]CommandEntry
	CommandQueues []CommandEntry
	//传输命令执行进度的通道
	CommandExecRetChan chan CommandExecRet

	services.UnimplementedRaftServiceServer
	/*----------------------------------------------------------------------------------------------------------------*/
	//接收投票的通道
	VoteChan chan *services.VoteResp
	//调整角色的通道
	ChangeRoleChan chan string
	//关闭通道
	Shutdown chan bool
	//是否已经进行过投票
	HasBeenVoted bool
	//所得到得选票数
	VoteCount int64
	//该 Raft Server 所投票的对象
	VoteFor string
	//选举超时时间节点
	ElectionTime time.Time
	//超过该变量限定的时间间隔，就会进入选举
	ElectionTimeout int
	/*----------------------------------------------------------------------------------------------------------------*/
	//心跳间隔
	Heartbeat int
	/*----------------------------------------------------------------------------------------------------------------*/
	//grpc 服务器
	GrpcServer *grpc.Server
	//grpc 监听器
	GrpcListener net.Listener
	//http 服务器
	HttpServer *gin.Engine
	//websocket upgrader
	WSUpgrader websocket.Upgrader
	//websocket 连接池
	WSConnPool map[string]*websocket.Conn
	/*----------------------------------------------------------------------------------------------------------------*/

	/*----------------------------------------------------------------------------------------------------------------*/
	Mutex sync.Mutex //访问raft资源时要用到的锁
	//   用于goroutine同步
	Wg sync.WaitGroup
	/*----------------------------------------------------------------------------------------------------------------*/

}

// Peer
// 当前 Raft 保存其他 Raft 节点的结构
type Peer struct {
	Addr         string
	Role         string
	LastLogIndex int64
}

// WorkerStatus
// 当前 Raft 保存 Worker 节点的结构
type WorkerInfo struct {
	Addr      string `json:"addr"`
	TickTok   time.Time
	UsedDisk  int `json:"usedDisk"`
	TotalDisk int `json:"totalDisk"`
	UsedMem   int `json:"usedMem"`
	TotalMem  int `json:"totalMem"`
}

// 配置信息
type RaftConfig struct {
	Active          bool   `yaml:"active"`
	Ip              string `yaml:"ip"`
	RpcPort         string `yaml:"rpc-port"`
	HttpPort        string `yaml:"http-port"`
	LogPath         string `yaml:"log-path"`
	MsgLevel        int    `yaml:"msg-level"`
	Heartbeat       int    `yaml:"heartbeat"`
	ElectionTimeout int    `yaml:"election-timeout"`
}

const (
	/* Raft 各个角色定义 */
	Follower  = "Follower"
	Candidate = "Candidate"
	Leader    = "Leader"
	Dead      = "Dead"
	HELP      = "HELP"
)

// 启动各 goroutine 和 rpc 服务器
// @receiver r
func (r *Raft) Start() {
	r.Wg.Add(3)
	go r.StartRaft()
	//TODO 启动 applier
	go r.StartGRPCServer()
	go r.StartHttpServer()
	r.Wg.Wait()
}

// var _ *Raft=
// Initialize
// 初始化Raft
func (r *Raft) Initialize() {
	//TODO 检查传入的配置文件的信息是否合法。
	r.Banner = "EDH"
	//初始化 Raft 服务基本信息
	//  根据配置文件application.yml中的配置，初始化当前Raft节点的地址
	r.Addr = r.Config.Ip + ":" + r.Config.RpcPort
	// 根据配置文件application.yml中的配置，初始化心跳间隔
	r.Heartbeat = r.Config.Heartbeat
	// 根据配置文件application.yml中的配置，初始化超时时间
	r.ElectionTimeout = r.Config.ElectionTimeout
	//  初始化存储Raft集群信息的变量，并将自己加进去
	r.Peers = make(map[string]Peer)
	//TODO 此处初始化时，可能需要读取本地的日志
	var p Peer
	p.Addr = r.Addr
	p.Role = r.Role
	p.LastLogIndex = 0
	r.Peers[r.Addr] = p
	//  初始化存储Worker状态信息的变量
	r.Workers = make(map[string]WorkerInfo)
	// 初始化日志变量，创建空的日志对象，
	// TODO，对于不是第一次启动的Raft节点，该方法有待优化
	r.Log = MakeEmptyLog()
	r.Msg = &Message{PrintLevel: r.Config.MsgLevel}
	// 初始化用于关闭进程的通道
	r.Shutdown = make(chan bool)
	//  初始化用于改变Raft节点角色的通道
	r.ChangeRoleChan = make(chan string)
	// 初始化命令队列
	r.CommandQueues = make([]CommandEntry, 0)
	//初始化Raft内部命令
	r.InitBuiltinCommand()
	//初始化 GRPC 服务对象
	listener, err := net.Listen("tcp", r.Addr)
	if err != nil {
		r.Msg.Error("%s grpc listen fail:%s", r.Addr, err)
		return
	}
	r.GrpcListener = listener
	r.GrpcServer = grpc.NewServer()
	//注册 RPC 方法
	services.RegisterRaftServiceServer(r.GrpcServer, r)
	// 创建websocket的upgrader
	r.WSUpgrader = websocket.Upgrader{
		CheckOrigin:      func(r *http.Request) bool { return true },
		HandshakeTimeout: time.Duration(time.Second * 10),
	}
	// 初始化websocket 连接池
	r.WSConnPool = make(map[string]*websocket.Conn)
	//初始化 HTTP 服务对象，配置跨域
	gin.SetMode(gin.ReleaseMode)
	r.HttpServer = gin.Default()
	r.HttpServer.SetTrustedProxies([]string{"127.0.0.1"})
	corsConfig := cors.DefaultConfig()
	corsConfig.AllowAllOrigins = true
	corsConfig.AllowMethods = []string{"GET", "POST", "PUT", "DELETE"}
	corsConfig.AllowHeaders = []string{"Accept", "Content-Type", "Content-Length", "Authorization"}
	r.HttpServer.Use(cors.New(corsConfig))
	//  注册 HTTP 方法
	r.InitBuiltinHttpServices()
}

// StartRaft
// 通过 goroutine 的方式启动 Raft，各节点开始以不同的角色运行
func (r *Raft) StartRaft() {
	defer r.Wg.Done()
	go r.StartTicker()
	//TODO go routine 检查 Command 的执行情况
	for r.Role != Dead {
		switch r.Role {
		case Leader:
			r.RunAsLeader()
		case Candidate:
			r.RunAsCandidate()
		case Follower:
			r.RunAsFollower()
		}
	}
}

// StartTicker
// 定时器，不同角色的定时任务有所不同。Leader->日志复制，Follower->选举时间（ElectionTime）超时时进入成为Candidate
func (r *Raft) StartTicker() {
	ticker := time.NewTicker(time.Second * 2)
	for r.Role != Dead {
		select {
		case <-r.Shutdown:
			r.Msg.Debug("%s ticker is stopped", r.Addr)
			return
		case <-ticker.C:
			if r.Role == Leader {
				r.Msg.Debug("%s heartbeat start", r.Addr)
				r.AppendEntries()
			}
			if time.Now().After(r.ElectionTime) {
				r.Msg.Debug("%s election time out", r.Addr)
				r.ChangeRoleChan <- Candidate
			}
		}
	}
}

// RunAsLeader
// 作为 Leader 运行
func (r *Raft) RunAsLeader() {
	r.Msg.Info("%s is running as leader", r.Addr)
	/*TODO 不一定要这个*/
	//go r.ExecuteCommand()
	for r.Role == Leader {
		select {
		case <-r.Shutdown:
			r.Role = Dead
			return
		case role := <-r.ChangeRoleChan:
			r.Role = role
		}
	}
}

//	作为 Candidate 运行
//
// @receiver r
func (r *Raft) RunAsCandidate() {
	r.Msg.Debug("%s is running as candidate", r.Addr)
	r.HasBeenVoted = false
	r.VoteCount = 0
	for r.Role == Candidate {
		if !r.HasBeenVoted {
			r.RequestVote()
			r.HasBeenVoted = true //标记该 Candidate 已经申请过投票
		}
		select {
		case <-r.Shutdown:
			r.Role = Dead
			return
		case v := <-r.VoteChan:
			if v.VoteYN {
				r.VoteCount++
			}
			if r.VoteCount >= int64(len(r.Peers)/2+1) {
				//选举成功后，广播日志，告诉其他节点自己已成为 Leader
				r.AppendEntries()
			}
		case role := <-r.ChangeRoleChan:
			r.Role = role
		}
	}
}

//	作为 Follower 运行
//
// @receiver r
func (r *Raft) RunAsFollower() {
	r.Msg.Debug("%s is running as follower", r.Addr)
	for r.Role == Follower {
		select {
		case <-r.Shutdown:
			r.Role = Dead
			return
		case role := <-r.ChangeRoleChan:
			r.Role = role
		}
	}
}

//	启动该节点的 GRPC 服务器
//
// @receiver r
func (r *Raft) StartGRPCServer() {
	defer r.Wg.Done()
	r.Msg.Info("grpc server running at %s", r.Addr)
	serveErr := r.GrpcServer.Serve(r.GrpcListener)
	if serveErr != nil {
		r.Msg.Error("grpc serve fail:%s", serveErr)
	}
}

func (r *Raft) StartHttpServer() {
	defer r.Wg.Done()
	http_addr := r.Config.Ip + ":" + r.Config.HttpPort
	r.Msg.Info("http server running at %s", http_addr)
	serveErr := r.HttpServer.Run(http_addr)
	if serveErr != nil {
		r.Msg.Error("http serve fail:%s", serveErr)
	}
}
